package com.learn.lb.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 使用updateStateByKey算子实现全局统计key出现的次数
 *
 * @author laibo
 * @since 2019/9/5 22:04
 *
 */
object UpdateStateByKeyDemo {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("WorkCountFileStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    //设置checkpoint目录，开启checkpoint机制
    //因为key的state是在内存维护的，如果宕机，则重启之后之前维护的状态就没有了，所以要长期保存它的话需要启用checkpoint，以便恢复数据
    ssc.checkpoint("E:\\idea\\workspace\\spark-learn\\src\\main\\resources\\checkpoint")
    val lines = ssc.socketTextStream("master", 9999)
    val result = lines.flatMap(_.split(" ")).map((_, 1))
//    val state = result.updateStateByKey[Int]((newValues, preValues)=>{
//      //newValues 新来的数据相同的key出现了多少次
//      //用新来的数据出现的次数加上以前出现的次数 就是全局出现的次数
//      val newCount = newValues.sum + preValues.getOrElse(0)
//      Some(newCount)
//    })
//    state.print()
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 把当前的数据去更新已有的数据
   * @param newValues 新有的状态
   * @param preValues 已有的状态
   * @return
   */
  def updateFunction(newValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
    //newValues 新来的数据相同的key出现了多少次
    //用新来的数据出现的次数加上以前出现的次数 就是全局出现的次数
    val newCount = newValues.sum + preValues.getOrElse(0)
    Some(newCount)
  }
}
